
package cfar.parallel.map;

import cfar.ff.data.Stream;
import cfar.ff.fun.F1;
import cfar.ff.fun.F2;
import cfar.ff.fun.F4;
import cfar.ff.fun.Fun;
import cfar.ff.fun.Seq;
import cfar.ff.interfaces.Sequence;

import java.util.Iterator;

/**
 *
 * @author badi
 */
public class RMap<S,A,B> implements F4< F1<Sequence<Stream,B>, B> , Sequence<Stream,Integer> , F1<A,B> , Sequence<S,A> , Sequence<S,B>> {

    public Sequence<S, B> f( F1<Sequence<Stream, B>, B> chooser
                           , Sequence<Stream, Integer>  redundancies
                           , F1<A, B>                   f
                           , final Sequence<S, A>       seq
                           )
    {

        Sequence<Stream, Sequence<Stream,A>> branches = branch_out(redundancies, seq);
        Sequence<Stream, Sequence<Stream,B>> apps = mapping(f, branches);
        Sequence<Stream, B> results = ParMap.map(chooser, apps);

        
        F1<B, Sequence<S,B>> maker = new F1<B, Sequence<S, B>>() {

            public Sequence<S, B> f(B b) {
                return (Sequence<S, B>) Fun.<S,A,B>pure().f(seq, b);
            }
        };

        F2<Sequence<S,B>, Sequence<S,B>, Sequence<S,B>> joiner = new F2<Sequence<S, B>, Sequence<S, B>, Sequence<S, B>>() {

            public Sequence<S, B> f(Sequence<S, B> s1, Sequence<S, B> s2) {
                return Seq.concat(s1, s2);
            }
        };


        return Seq.parFoldl1(joiner, ParMap.map(maker, results));
    }

    private Sequence<Stream, Sequence<Stream,A>> branch_out ( Sequence<Stream,Integer> redundancies
                                                            , Sequence<S,A> seq
                                                            )
    {
        Sequence<Stream, Sequence<Stream,A>> res = Stream.nil();

        Iterator<Integer> reds = redundancies.iterator();
        Iterator<A> vals = seq.iterator();

        while(reds.hasNext() && vals.hasNext()){
            Sequence<Stream,A> r = Stream.<A>nil();
            int count = reds.next();
            A val = vals.next();
            for(int i = 0; i < count; i++)
                r = r.snoc(val);
            res = res.snoc(r);
        }

        return res;
    }

    private Sequence<Stream, Sequence<Stream,B>> mapping ( final F1<A,B> f
                                                         , Sequence<Stream, Sequence<Stream,A>> branches
                                                         )
    {
        F1<Sequence<Stream,A>, Sequence<Stream,B>> mapper = new F1<Sequence<Stream, A>, Sequence<Stream, B>>() {

            public Sequence<Stream, B> f(Sequence<Stream, A> s) {
                return ParMap.map(f, s);
            }
        };

        return ParMap.map(mapper, branches);
    }
}
